//package com.example.demo.config;
//
//
//import com.example.demo.constant.CacheConstant;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.context.annotation.DependsOn;
//import org.springframework.data.redis.connection.RedisConnectionFactory;
//import org.springframework.data.redis.connection.stream.Consumer;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.ReadOffset;
//import org.springframework.data.redis.connection.stream.StreamOffset;
//import org.springframework.data.redis.stream.StreamMessageListenerContainer;
//
//import java.time.Duration;
//
///**
// * @author ：zz
// * @date ：Created in 2022/5/19 10:23
// * @description：消费组配置
// */
//@Configuration
//public class RedisStreamConfiguration {
//
//
//    @Bean(initMethod = "start", destroyMethod = "stop")
//    @DependsOn("redisConnectionFactory")
//    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer(
//            RedisConnectionFactory factory,
//            ConsumeListenerFdelay fdelayListener,     // 你的监听器
//            ConsumeListenerFjitter fjitterListener,
//            ConsumeListenerFthroughput fthroughputListener
//    ) {
//        if (fdelayListener == null || fjitterListener == null || fthroughputListener == null) {
//            throw new IllegalArgumentException("All listeners must be initialized.");
//        }
//
//        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
//                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
//                        .pollTimeout(Duration.ofSeconds(1))
//                        .build();
//
//        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
//                StreamMessageListenerContainer.create(factory);
//
//        try {
//            // FDELAY监听
//            container.receive(
//                    Consumer.from("group-fdelay-1", "consumer-1"),
//                    StreamOffset.create(CacheConstant.REDIS_STEAM_01, ReadOffset.lastConsumed()),
//                    fdelayListener);
//
//            // FJITTER监听
//            container.receive(
//                    Consumer.from("group-fjitter", "consumer-1"),
//                    StreamOffset.create(CacheConstant.REDIS_STEAM_02, ReadOffset.lastConsumed()),
//                    fjitterListener);
//
//            // FTHROUGHPUT监听
//            container.receive(
//                    Consumer.from("group-fthroughput", "consumer-1"),
//                    StreamOffset.create(CacheConstant.REDIS_STEAM_03, ReadOffset.lastConsumed()),
//                    fthroughputListener);
//        } catch (Exception e) {
//            // Log the exception and rethrow it to prevent the container from starting incorrectly
//            System.err.println("Failed to initialize message listeners: " + e.getMessage());
//            throw e;
//        }
//
//        return container;
//    }
//
//}
